package com.atguigu.dga.governance.service.impl;

import com.atguigu.dga.common.SpringBeanProvider;
import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.ds.service.TDsTaskDefinitionService;
import com.atguigu.dga.ds.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.service.GovernanceAssessDetailService;
import com.atguigu.dga.governance.service.GovernanceMetricService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Sets;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2023-06-30
 */
@Service
@DS("dga")
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {


    ThreadPoolExecutor poolExecutor=  new ThreadPoolExecutor(40,80,1L, TimeUnit.MINUTES,new LinkedBlockingDeque<>(2000));


    @Autowired
    TableMetaInfoService tableMetaInfoService;

    @Autowired
    GovernanceMetricService governanceMetricService;

    @Autowired
    SpringBeanProvider springBeanProvider;

    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;

    public  void  mainAssess(String assessDate  ){
        // 0 幂等处理  把考评当日已存在的明细清除
        remove(new QueryWrapper<GovernanceAssessDetail>().eq("assess_date" ,assessDate));

        //1 考评准备
        //1.1  查询所有的元数据表+辅助信息表
        List<TableMetaInfo> tableMetaInfoList=   tableMetaInfoService.getTableMetaInfoAllList(assessDate);
        Map<String ,TableMetaInfo>  tableMetaInfoMap=new HashMap<>();
        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            tableMetaInfoMap.put(tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName(),tableMetaInfo);
        }
          Map<String, TableMetaInfo> map1 = tableMetaInfoList.stream().collect(java.util.stream.Collectors.toMap(tableMetaInfo->tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName(), tableMetaInfo -> tableMetaInfo));
        // Map<String, List<TDsTaskInstance>> map = tableMetaInfoList.stream().collect(Collectors.groupingBy(TDsTaskInstance::getName));


        System.out.println("tableMetaInfoList = " + tableMetaInfoList);
        //1.2  查询所有的指标信息
        List<GovernanceMetric> governanceMetricList = governanceMetricService.list(new QueryWrapper<GovernanceMetric>().eq("is_disabled", "0"));

        //1.3  ds信息（任务定义 、 任务的实例 )
        //  哪些任务的实例  考评当日   涉及考评的元数据表    tableMetaInfoList
        List<TDsTaskInstance> tDsTaskInstanceList= tDsTaskInstanceService.getTaskInstanceListForAssess( assessDate, tableMetaInfoList) ;
        Map<String,List<TDsTaskInstance>> tDsTaskInstanceMap=new HashMap<>(tDsTaskInstanceList.size()*2);
        for (TDsTaskInstance tDsTaskInstance : tDsTaskInstanceList) {
            //查询任务名称在map中是否有list
            // 如果没有list 创建list 把任务实例写入list
            //   把任务实例写入list
            List<TDsTaskInstance> curTaskInstanceList = tDsTaskInstanceMap.get(tDsTaskInstance.getName());
            if (curTaskInstanceList==null){
                curTaskInstanceList = new ArrayList<>();
                tDsTaskInstanceMap.put(tDsTaskInstance.getName(),curTaskInstanceList);
            }
            curTaskInstanceList.add(tDsTaskInstance);
        }

        // 哪些任务定义：  考评当日运行过的任务实例的任务定义   用实例id 可以关联定义
          List<TDsTaskDefinition> tDsTaskDefinitionList=  tDsTaskDefinitionService.getTaskDefinitionListForAssess(tDsTaskInstanceList) ;
          Map<String,TDsTaskDefinition> tDsTaskDefinitionMap= new HashMap<>(tDsTaskInstanceList.size()*2);
          for (TDsTaskDefinition tDsTaskDefinition : tDsTaskDefinitionList) {
             tDsTaskDefinitionMap.put(tDsTaskDefinition.getName(),tDsTaskDefinition);
          }

        long start = System.currentTimeMillis();
        //2 根据表 指标 进行逐一考评   两层循环 进行考评 -->考评明细的列表List<GovernanceAssessDetail>
        List<GovernanceAssessDetail> governanceAssessDetailList=new ArrayList<>(tableMetaInfoList.size()*governanceMetricList.size());
        List<CompletableFuture<GovernanceAssessDetail>> governanceAssessDetailFutureList=new ArrayList<>(tableMetaInfoList.size()*governanceMetricList.size());
        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            for (GovernanceMetric governanceMetric : governanceMetricList) {
                //跳过 白名单
                String skipAssessTables = governanceMetric.getSkipAssessTables();
                if(skipAssessTables!=null&&skipAssessTables.length()>0) {
                    String[] skipTableArr = skipAssessTables.split(",");
                    HashSet<String> skipTableSet = Sets.newHashSet(skipTableArr);
                    if (skipTableSet.contains(tableMetaInfo.getTableName())) {
                        continue;
                    }
                }

                AssessParam assessParam = new AssessParam();
                assessParam.setAssessDate(assessDate);
                assessParam.setGovernanceMetric(governanceMetric);
                assessParam.setTableMetaInfo(tableMetaInfo);
                assessParam.setTableMetaInfoAllList(tableMetaInfoList);
                assessParam.setTableMetaInfoAllMap(tableMetaInfoMap);

                String taskName = tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName();
                TDsTaskDefinition tDsTaskDefinition = tDsTaskDefinitionMap.get(taskName);
                assessParam.setTDsTaskDefinition(tDsTaskDefinition);

                List<TDsTaskInstance> curTaskDefiniton = tDsTaskInstanceMap.get(taskName);
                assessParam.setTDsTaskInstanceList(curTaskDefiniton);
                // metricAssess(assessParam);
//                if(tableMetaInfo.getTableMetaInfoExtra().getDwLevel().equals("DWD")){
//                    System.out.println(111);
//                }
                String metricCode = governanceMetric.getMetricCode();
                Assessor assessor = springBeanProvider.getBeanByName(metricCode, Assessor.class);
                CompletableFuture<GovernanceAssessDetail> future=CompletableFuture.supplyAsync(()->{
                    return assessor.assess(assessParam);
                },poolExecutor);  //获得期货

               // GovernanceAssessDetail governanceAssessDetail = assessor.assess(assessParam); //现货交割
                governanceAssessDetailFutureList.add(future);
               // governanceAssessDetailList.add(governanceAssessDetail);
                //System.out.println("governanceAssessDetail = " + governanceAssessDetail);
            }
        }

        //期货换现货
        governanceAssessDetailList= governanceAssessDetailFutureList.stream().map(future->future.join()).collect(Collectors.toList());

        System.err.println("duration = " + (System.currentTimeMillis()-start));

        //3  保存考评结果
        saveOrUpdateBatch(governanceAssessDetailList);


    }

    private void metricAssess(AssessParam assessParam) {
        if(assessParam.getGovernanceMetric().equals("xxx")){
            //x
        }else if (assessParam.getGovernanceMetric().equals("xxx")){
            ///xxx
        }
    }
}
